package com.poetic.network.flow;

import com.poetic.network.flow.domain.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-13.
 * </pre>
 *
 * @author lianghuikun
 */
public class WindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String path = "/Users/lianghuikun/indigo/poetic/flink/flink-demo/src/main/resources/sensor.txt";
//        DataStreamSource<String> stream = env.readTextFile(path);
        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);
        stream.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String data) throws Exception {
                String[] dataArr = data.split("\\,");
                return new SensorReading(dataArr[0].trim(), Long.valueOf(dataArr[1].trim()), Double.valueOf(dataArr[2].trim()));
            }
        }).keyBy(new KeySelector<SensorReading, String>() {
            @Override
            public String getKey(SensorReading value) throws Exception {
                return value.getId();
            }
        }).process(new KeyedProcessFunction<String, SensorReading, String>() {
            @Override
            public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
                // 注册定时器-事件时间
//                ctx.timerService().registerEventTimeTimer();
                // 注册定时器-处理时间
//                ctx.timerService().registerProcessingTimeTimer();
                // 水位线，也就是当前事件时间
//                ctx.timerService().currentWatermark()
                 // 当前处理事件时间
//                ctx.timerService().currentProcessingTime()
            }
        });
        env.execute("window test");

    }
}
